package com.ada.spark.partitioner

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
  * Spark目前支持Hash分区和Range分区，用户也可以自定义分区，Hash分区为当前的默认分区，Spark中分区器直接决定了RDD中分区的个数、RDD中每条数据经过Shuffle过程属于哪个分区和Reduce的个数
  *
  * 注意：
  * (1) 只有Key-Value类型的RDD才有分区器的，非Key-Value类型的RDD分区器的值是None
  * (2) 每个RDD的分区ID范围：0~numPartitions-1，决定这个值是属于那个分区的。
  *
  * HashPartitioner分区的原理：对于给定的key，计算其hashCode，并除以分区的个数取余，如果余数小于0，则用余数+分区的个数（否则加0），最后返回的值就是这个key所属的分区ID。
  * HashPartitioner分区弊端：可能导致每个分区中数据量的不均匀，极端情况下会导致某些分区拥有RDD的全部数据。
  *
  * RangePartitioner作用：将一定范围内的数映射到某一个分区内，尽量保证每个分区中数据量的均匀，而且分区与分区之间是有序的，一个分区中的元素肯定都是比另一个分区内的元素小或者大，但是分区内的元素是不能保证顺序的。简单的说就是将一定范围内的数映射到某一个分区内。实现过程为：
  * 第一步：先重整个RDD中抽取出样本数据，将样本数据排序，计算出每个分区的最大key值，形成一个Array[KEY]类型的数组变量rangeBounds；
  * 第二步：判断key在rangeBounds中所处的范围，给出该key值在下一个RDD中的分区id下标；该分区器要求RDD中的KEY类型必须是可以排序的
  */
object HashPartitionerTest {
    def main(args: Array[String]): Unit = {
        //初始化配置信息及SparkContext
        val sparkConf: SparkConf = new SparkConf().setAppName("HashPartitionerTest").setMaster("local[*]")
        val sc = new SparkContext(sparkConf)

        val pairs = sc.parallelize(List((1, 1), (2, 2), (3, 3)))
        println(pairs.partitioner.size) //0

        import org.apache.spark.HashPartitioner
        val partitioned = pairs.partitionBy(new HashPartitioner(2))
        println(pairs.partitioner.size) //0

        val nopar = sc.parallelize(List((1, 3), (1, 2), (2, 4), (2, 3), (3, 6), (3, 8)), 8)

        val result: RDD[String] = nopar.mapPartitionsWithIndex((index, iter) => {
            Iterator(index.toString + " : " + iter.mkString("|"))
        })
        println(result.collect().mkString(","));
        //0 : ,1 : (1,3),2 : (1,2),3 : (2,4),4 : ,5 : (2,3),6 : (3,6),7 : (3,8)

        val hashpar = nopar.partitionBy(new HashPartitioner(7))

        println(hashpar.count) //6
        println(hashpar.partitioner.size) //1

        val result2: RDD[Int] = hashpar.mapPartitions(iter => Iterator(iter.length))
        println(result2.collect().mkString(",")); //0,2,2,2,0,0,0
    }
}
